package com.example.mqttmodule.publish;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.extern.log4j.Log4j2;
import org.dromara.mica.mqtt.spring.client.MqttClientTemplate;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

@Service
@Log4j2
public class MqttPublisher {
    private final MessageChannel mqttOutboundChannel;

    @Autowired
    private MqttClientTemplate client;

    public Gson gson = new GsonBuilder().create();

    @Value("${mqtt.topics.status}")
    String topic;

    @Value("${mqtt.qos}")
    private int qos;

    @Autowired
    public MqttPublisher(@Qualifier("mqttOutboundChannel") MessageChannel channel) {
        this.mqttOutboundChannel = channel;
    }

    /**
     * 发布
     * @param retained
     *                true（保留消息）
     *                   服务端（Broker）会永久存储这条消息（直到被新的保留消息覆盖）。
     *                   当有新客户端订阅该主题（Topic）时，服务端会立即推送这条保留消息给新订阅者。
     *
     *                    典型应用场景：
     *                       设备上线时获取最后一次状态（如传感器最新数据）。
     *                       发布系统默认配置。
     *
     *                false（不保留消息）
     *                   服务端不会存储该消息，仅分发给当前已订阅的客户端。
     *                   新订阅者不会收到这条消息（除非发布者再次发送）。
     *
     * @param qos     对消息处理的几种机制。
     *                0 表示的是订阅者没收到消息不会再次发送，消息会丢失。
     *                1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。
     *                2 多了一次去重的动作，确保订阅者收到的消息有一次。
     * @param retained
     */
    public void sendMessage(String message) throws Exception {
        this.sendMessage(topic,message);
    }

    public void sendMessage(String topic,String message) throws Exception {
        try {
            Message msg = MessageBuilder.withPayload(message)
                    .setHeader(MqttHeaders.TOPIC, topic)
                    .setHeader(MqttHeaders.QOS, 1)
                    .setHeader(MqttHeaders.RETAINED, false).build();
            boolean sent = mqttOutboundChannel.send(msg, 6000 // 超时时间(ms)
            );
            if (!sent) {
                throw new Exception("Message send timeout");
            }else {
                log.info(topic + " Message sent: " + message);
            }
        }catch (MqttSecurityException e){
            log.error("Failed to send message", e);
            throw new Exception("Failed to send message" + e.getMessage(),e);
        }catch (Exception e) {
            log.error("Failed to send message", e);
            throw new Exception("Failed to send message" + e.getMessage(),e);
        }
    }

    public boolean publish(String message) {
        client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
        return true;
    }

    public boolean subscribe() {
        client.subQos0("/test/#", (context, topic, message, payload) -> {
            log.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
        });
        return true;
    }

}